package com.aotain.jupiter.sqlinjection;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import com.aotain.common.CommonFunction;


public class SQLInjectionMap implements MapFunction<Tuple2<String,String>,Tuple2<String,SQLInjectionTuple>>{

	private String injectionKeyList = "";
	
	public SQLInjectionMap(String injectionkeylist) {
		injectionKeyList = injectionkeylist;
	}
	
	/**
	 * 
	 */
	private static final long serialVersionUID = 6538895754234752921L;

	@Override
	public Tuple2<String, SQLInjectionTuple> map(Tuple2<String, String> arg0)
			throws Exception {
		// TODO Auto-generated method stub
//		//dip+ "|" +dport+ "|" +sip+ "|" +sport+ "|" +gisstrBase64+ "|" 
//		//+ domainname+ "|" +url+ "|" +type+ "|" +delaytime+ "|" +idcno+ "|" 
//		// +upStreamOctets+ "|" +upStreamPacket+ "|" +status+ "|" +userAgent+ "|" +cookie+ "|" 
//		// +dnStreamOctets+ "|" +dnStreamPacket+ "|" +refer
		
		Tuple2<String, SQLInjectionTuple> ret = new Tuple2<String, SQLInjectionTuple>();
		String logString = arg0.f1; //格式化后的原始数据
		String[] items = logString.split("\\|",-1);
		if(items.length <= 10)
			return ret;
		String sip = arg0.f0;
		String dip = items[0];
		String gis = items[4];
		String idc = items[9];
		String dport = items[1];
		String sport = items[3];
		String type = items[7];
		
		String url = items[6];
		
		if(url.isEmpty())
			return ret;
		
		if(CommonFunction.MatchSQLInjection(url,injectionKeyList) >= 2){
			String Key = dip + "|" + dport + "|" + sip + "|" + type ;
			SQLInjectionTuple tuple = new SQLInjectionTuple();
			tuple.setDestIP(dip);
			tuple.setSourceIP(sip);
			tuple.setGIS(gis);
			tuple.setIDC(idc);
			tuple.setDestPort(dport);
			tuple.setSourcePort(sport);
			tuple.setType(type);
			tuple.setURL(url);
			tuple.setCounter(1);
			
			long upStreamOctets = 0; //UPSTREAMOCTETS
			long upStreamPacket = 0; //UPSTREAMPACKET			
			upStreamOctets = Long.parseLong(items[10]);
			upStreamPacket = Long.parseLong(items[11]);
			tuple.setUpStreamOctets(upStreamOctets);
			tuple.setUpStreamPacket(upStreamPacket);
			
			ret.setFields(Key, tuple);
		}
		return ret;
	}

}
